package cn.doitedu

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

import scala.collection.mutable.ListBuffer


case class UserEvent(guid: Long, event_id: String, action_time: Long)

case class AttrRes(model_id: String, target_event: String, attr_strategy: String, guid: Long, target_event_time: Long, attr_event: String, attr_event_time: Long, weight: Float)


object Job03_归因分析dws表计算 {

  def main(args: Array[String]): Unit = {

    if(args.length<3){
      println(
        """
          |
          |参数不正确：
          |args(0): 模型id
          |args(1): 目标事件id
          |args(2): 待归因'事件id'（逗号连接）
          |
          |
          |""".stripMargin)
    }

    val modelId = args(0)
    val targetEventId = args(1)
    // 'e1','e2','e3'
    val attrEventIds = args(2)

    val spark = SparkSession.builder()
      .appName("归因分析")
      .master("local")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._


    // 目标事件: e4
    // 待归因事件: 'e1','e2','e3'
    val df = spark.sql(
      s"""
         |
         |select
         |    guid,
         |    event_id,
         |    action_time
         |from tmp.user_action_detail
         |where dt='2024-04-25'
         |and event_id in ('${targetEventId}',${attrEventIds})
         |
         |""".stripMargin)

    val rdd: RDD[UserEvent] = df.as[UserEvent].rdd


    // 按照 guid 分组，收集一个人的模型相关的所有事件
    val resultRdd: RDD[AttrRes] = rdd.groupBy(_.guid)
      .map(_._2)
      .flatMap(iter => {

        // 创建一个接收计算结果的List
        val results = ListBuffer.empty[AttrRes]


        // 拿到了一组用户的行为,对数据按时间先后排序
        // List(UserEvent(1,e3,10), UserEvent(1,e1,11), UserEvent(1,e2,12), UserEvent(1,e3,13), UserEvent(1,e4,14), UserEvent(1,e2,15), UserEvent(1,e1,16), UserEvent(1,e3,17), UserEvent(1,e4,18), UserEvent(1,e1,19))
        val lst = iter.toList.sortBy(_.action_time)
        println("原始行为序列： " + lst)

        // 去掉最后一个目标事件之后的事件
        val lastTargetEventIndex = lst.lastIndexWhere(_.event_id == targetEventId)
        // List(UserEvent(1,e3,10), UserEvent(1,e1,11), UserEvent(1,e2,12), UserEvent(1,e3,13), UserEvent(1,e4,14), UserEvent(1,e2,15), UserEvent(1,e1,16), UserEvent(1,e3,17), UserEvent(1,e4,18))
        val slicedEvents = lst.take(lastTargetEventIndex + 1)
        println("截掉多余事件的序列 ： " + slicedEvents)

        // 遇到一个目标事件就把行为序列分一段 ,eventSubList 用来装一段事件
        val eventSubList = ListBuffer.empty[UserEvent]
        // 遍历用户的行为序列，开始分段
        for (e <- slicedEvents) {

          // 如果遍历到的事件不是目标事件，则存入分段中
          if (!e.event_id.equals(targetEventId)) {
            eventSubList += e
          } else {
            // 否则，说明一段已完成
            // 调用各种策略算法，对这一段事件取做归因计算
            println("准备计算分段：  " + eventSubList)

            results += firstAttr(eventSubList, e.action_time, modelId, targetEventId) // 首次触点归因策略
            results += lastAttr(eventSubList, e.action_time, modelId, targetEventId) // 末次触点归因策略
            results ++= lineAttr(eventSubList, e.action_time, modelId, targetEventId) // 线性归因策略
            results ++= locationAttr(eventSubList, e.action_time, modelId, targetEventId) // 位置归因策略
            results ++= timeDecayAttr(eventSubList, e.action_time, modelId, targetEventId) // 时间衰减归因策略

            // 算完后，清空list，给下一段做准备
            eventSubList.clear()

          }
        }
        // 返回最终结果
        results

      })

    // resultRdd.toDF().write.mode("overwrite").saveAsTable("tmp.user_action_attr")

    resultRdd.toDF().show(100)



    spark.stop()

  }


  def firstAttr(eventList: ListBuffer[UserEvent], targetEventTime: Long, modelId: String, targetEventId: String): AttrRes = {
    //  ListBuffer(UserEvent(1,e3,10), UserEvent(1,e1,11), UserEvent(1,e2,12), UserEvent(1,e3,13))
    val firstEvent = eventList.head
    AttrRes(modelId, targetEventId, "首次触点归因策略", firstEvent.guid, targetEventTime, firstEvent.event_id, firstEvent.action_time, 1.0f)
  }


  def lastAttr(eventList: ListBuffer[UserEvent], targetEventTime: Long, modelId: String, targetEventId: String): AttrRes = {
    //  ListBuffer(UserEvent(1,e3,10), UserEvent(1,e1,11), UserEvent(1,e2,12), UserEvent(1,e3,13))
    val lastEvent = eventList.last
    AttrRes(modelId, targetEventId, "末次触点归因策略", lastEvent.guid, targetEventTime, lastEvent.event_id, lastEvent.action_time, 1.0f)
  }


  def lineAttr(eventList: ListBuffer[UserEvent], targetEventTime: Long, modelId: String, targetEventId: String): List[AttrRes] = {

    //  ListBuffer(UserEvent(1,e3,10), UserEvent(1,e1,11), UserEvent(1,e2,12), UserEvent(1,e3,13))
    val tmp = for (elem <- eventList) yield {
      AttrRes(modelId, targetEventId, "线性归因策略", elem.guid, targetEventTime, elem.event_id, elem.action_time, 1.0f / eventList.size)
    }

    tmp.toList

  }


  def locationAttr(eventList: ListBuffer[UserEvent], targetEventTime: Long, modelId: String, targetEventId: String): List[AttrRes] = {

    val res = ListBuffer.empty[AttrRes]

    //  ListBuffer(UserEvent(1,e3,10), UserEvent(1,e1,11), UserEvent(1,e2,12), UserEvent(1,e3,13))

    val first = eventList.head
    res += AttrRes(modelId,targetEventId,"位置归因策略",first.guid,targetEventTime,first.event_id,first.action_time,0.4f)

    val last = eventList.last
    res += AttrRes(modelId,targetEventId,"位置归因策略",last.guid,targetEventTime,last.event_id,last.action_time,0.4f)


    // 对中间的事件计算结果
    val middleEvents = eventList.slice(1, eventList.size - 1)
    if(middleEvents.length>0){

      val tmp: ListBuffer[AttrRes] = for (elem <- middleEvents) yield {
        AttrRes(modelId, targetEventId, "位置归因策略", elem.guid, targetEventTime, elem.event_id, elem.action_time, 0.2f / middleEvents.size)
      }
      res ++= tmp.toList
    }

    res.toList

  }


  // 时间衰减归因
  def timeDecayAttr(eventList: ListBuffer[UserEvent], targetEventTime: Long, modelId: String, targetEventId: String): List[AttrRes] = {

    // 求脚标之和
    val idxSum = eventList.indices.sum  + eventList.size


    val tmp = for (i <- eventList.indices) yield {
      val elem = eventList(i)
      AttrRes(modelId, targetEventId, "时间衰减归因策略", elem.guid, targetEventTime, elem.event_id, elem.action_time, (i+1.0f) / idxSum)

    }

    tmp.toList

  }


}
